package com.xl.flinkdemo.flinkState;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Collections;
import java.util.List;

/**
 * @Description: 算子状态
 * @Author: hxl
 * @Date: 2021/5/24 15:02
 * @param null:
 * @return: null
 **/
public class FlinkState {

  public static void main(String[] args) {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setParallelism(1);

    DataStream<String> stringDataStream =
        env.readTextFile("D:\\studyspace\\flinkdemo\\src\\main\\resources\\sensor.txt");


    // 定义一个有状态的map操作，实现一个count的功能，统计当前分区数据个数
    DataStream<Integer> mapStream = stringDataStream.map(new myMapFunction());
  }

  public static class myMapFunction implements MapFunction<String,Integer>, ListCheckpointed<Integer> {
    private Integer count = 0;
    @Override
    public Integer map(String s) throws Exception {
      return s.length();
    }

    @Override
    public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception {
      return Collections.singletonList(count);
    }

    //发生故障的时候从这里面保存的取出来进行恢复
    @Override
    public void restoreState(List<Integer> state) throws Exception {
      for (Integer num: state) {
        count += num;
      }
    }
  }
}
